package edu.zju.gis.dldsj.service.impl;

import edu.zju.gis.dldsj.dao.*;
import edu.zju.gis.dldsj.entity.*;
import edu.zju.gis.dldsj.service.WfService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author yanlong_lee@qq.com
 * @version 1.0 2018/11/06
 */
@Service
public class WfServiceImpl implements WfService {
    @Autowired
    private AirflowDagMapper airflowDagMapper;
    @Autowired
    private AirflowDagStatsMapper airflowDagStatsMapper;
    @Autowired
    private AirflowDagRunMapper airflowDagRunMapper;
    @Autowired
    private DagTaskInstanceMapper taskInstanceMapper;
    @Autowired
    private DagTaskFailMapper taskFailMapper;
    @Autowired
    private WfDagMapper dagMapper;
    @Autowired
    private WfRunMapper runMapper;
    @Autowired
    private WfInstanceMapper instanceMapper;

    @Override
    public int insert(AirflowDag airflowDag) {
        return airflowDagMapper.insertSelective(airflowDag);
    }

    @Override
    public int insert(WfDag wfDag) {
        return dagMapper.insertSelective(wfDag);
    }

    @Transactional
    @Override
    public void insert(AirflowDag airflowDag, WfDag wfDag) {
        airflowDagMapper.insert(airflowDag);
        dagMapper.insert(wfDag);
    }

    @Override
    public void update(WfDag wfDag) {
        dagMapper.updateByPrimaryKeySelective(wfDag);
    }

    @Override
    public void deleteWfDag(String dagId) {
        dagMapper.deleteByPrimaryKey(dagId);
    }

    @Override
    public int insert(WfRun dagRun) {
        return runMapper.insertSelective(dagRun);
    }

    @Override
    public void update(WfRun dagRun) {
        runMapper.updateByPrimaryKeySelective(dagRun);
    }

    @Override
    public void deleteWfRun(int id) {
        runMapper.deleteByPrimaryKey(id);
    }

    @Override
    public int insert(WfInstance instance) {
        return instanceMapper.insertSelective(instance);
    }

    @Override
    public void update(WfInstance instance) {
        if (instance.getStartDate() == null && instance.getEndDate() == null && instance.getDuration() == null && instance.getState() == null)
            return;
        instanceMapper.updateByPrimaryKeySelective(instance);
    }

    @Override
    public void deleteWfInstance(String dagId, Date executionDate) {
        instanceMapper.deleteByDagRun(dagId, executionDate);
    }

    @Override
    public void deleteWfInstance(String dagId, Date executionDate, String taskId) {
        instanceMapper.deleteByPrimaryKey(dagId, executionDate, taskId);
    }

    @Override
    public int getDagCount() {
        return airflowDagMapper.selectCount();
    }

    @Override
    public AirflowDag getDag(String dagId) {
        return airflowDagMapper.selectByPrimaryKey(dagId);
    }

    @Override
    public List<AirflowDag> getDag(int offset, int size) {
        return airflowDagMapper.selectByPage(offset, size);
    }

    @Override
    public List<AirflowDagStats> getDagStats(String dagId) {
        return airflowDagStatsMapper.selectByDagId(dagId);
    }

    @Override
    public AirflowDagStats getDagStats(String dagId, DagState state) {
        return airflowDagStatsMapper.selectByPrimaryKey(dagId, state.getState());
    }

    @Override
    public List<AirflowDagRun> getDagRun(String dagId) {
        return airflowDagRunMapper.selectByDagId(dagId);
    }

    @Override
    public List<AirflowDagRun> getDagRun(DagState state) {
        return airflowDagRunMapper.selectByState(state.getState());
    }

    @Override
    public AirflowDagRun getDagRun(int id) {
        return airflowDagRunMapper.selectByPrimaryKey(id);
    }

    @Override
    public AirflowTaskInstance getTaskInstance(String dagId, String taskId, Date executionDate) {
        return taskInstanceMapper.selectByDagAndTask(dagId, taskId).stream()
                .filter(o -> o.getExecutionDate().equals(executionDate)).findFirst().orElse(null);
        //airflow 时间字段精确到微秒，读取到java中会损失精度，故不采用直接操作mapper的方式实现
    }

    @Override
    public List<AirflowTaskInstance> getTaskInstance(String dagId, Date executionDate) {
        return taskInstanceMapper.selectByDag(dagId).stream()
                .filter(o -> o.getExecutionDate().equals(executionDate)).collect(Collectors.toList());
    }

    @Override
    public List<AirflowTaskFail> getTaskFail(String dagId, Date executionDate) {
        return taskFailMapper.selectByDag(dagId).stream()
                .filter(o -> o.getExecutionDate().equals(executionDate)).collect(Collectors.toList());
    }

    @Override
    public WfDag getWfDag(String dagId) {
        return dagMapper.selectByPrimaryKey(dagId);
    }

    @Override
    public List<WfDag> getWfDag(String userId, int offset, int size) {
        return dagMapper.selectByPage(userId, offset, size);
    }

    @Override
    public WfRun getWfRun(int id) {
        return runMapper.selectByPrimaryKey(id);
    }

    @Override
    public List<WfRun> getWfRun(String dagId) {
        return runMapper.selectByDag(dagId);
    }

    @Override
    public List<WfRun> getWfRun(DagTaskState state) {
        return runMapper.selectByState(state.getState());
    }

    @Override
    public List<WfRun> getWfRun(String dagId, DagTaskState state) {
        return runMapper.selectByDagAndState(dagId, state.getState());
    }

    @Override
    public List<WfInstance> getWfInstance(String dagId, Date executionDate) {
        return instanceMapper.selectByDagAndTime(dagId, executionDate);
    }

    @Override
    public WfInstance getWfInstance(String dagId, Date executionDate, String taskId) {
        return instanceMapper.selectByPrimaryKey(dagId, executionDate, taskId);
    }
}
